package com.yichat.micro.msghttpservice.core.record.batch;

import com.weibo.api.motan.util.ConcurrentHashSet;
import com.yichat.micro.msghttpservice.common.util.JsonUtil;
import com.yichat.micro.msghttpservice.core.record.mapper.MultiChildRecordMapper;
import com.yichat.micro.msghttpservice.core.record.model.MultiChildRecord;
import com.yichat.micro.msghttpservice.core.record.model.MultiChildRecordExample;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

/**
 * Created by chenmohan on 2019/04/25
 *
 * @description 批量请求的日志批量处理
 */
@Slf4j
@Component
public class MultiChildRecordBatch {

    @Value("${msg-http-service.multiRecord.batch-insert.capacity}")
    Integer insertCapacity;

    @Autowired
    MultiChildRecordMapper multiRecordMapper;

    protected BlockingQueue<MultiChildRecord> insertQueue;

    protected BlockingQueue<MultiChildRecord> updateQueue;

    protected MultiChildRecordExample example = new MultiChildRecordExample();

    @PostConstruct
    public void init() {
        insertQueue = new ArrayBlockingQueue<>(insertCapacity);
        updateQueue = new ArrayBlockingQueue<>(insertCapacity);
    }


    /**
     * 插入中的record对象id,在批量入库后会删除对应的id
     */
    protected ConcurrentHashSet<String> insertingMultiRecordSet = new ConcurrentHashSet<>();

    private String buildSetKey(MultiChildRecord record) {
        return record.getRecordId() + "-" + record.getCurrentIndex();
    }

    public void insert(MultiChildRecord record) {
        insertQueue.add(record);
        insertingMultiRecordSet.add(buildSetKey(record));
    }

    public void update(MultiChildRecord record) {
        updateQueue.add(record);
    }

    @Scheduled(cron = "0/5 * *  * * ? ")
    public void batch() {
        insert();
        //每次新增请求日志完成后再执行更新请求日志操作
        update();
    }

    private void insert() {
        List<MultiChildRecord> batchInsertList = new ArrayList<>(100);
        Set<String> removingIdSet = new HashSet<>(100);
        //一次最多写入5000条请求日志
        insertQueue.drainTo(batchInsertList, 5000);
        if (batchInsertList.size() == 0) {
            return;
        }
        try {
            multiRecordMapper.batchInsert(batchInsertList);
            removingIdSet = batchInsertList.parallelStream().map(e -> buildSetKey(e)).collect(Collectors.toSet());
            insertingMultiRecordSet.removeAll(removingIdSet);
            log.info("id={}【子】请求入库", Arrays.toString(removingIdSet.toArray()));
        } catch (Exception e) {
            log.error("批量插入MultiChildRecord异常", e);
            for (MultiChildRecord r : batchInsertList) {
                try {
                    multiRecordMapper.insertSelective(r);
                    removingIdSet.add(buildSetKey(r));
                } catch (Exception e2) {
                    log.error("单条请求插入MultiChildRecord异常 , info => {}", JsonUtil.toJSON(r), e2);
                }
            }
        }
        //删除
        insertingMultiRecordSet.removeAll(removingIdSet);
    }

    private void update() {
        List<MultiChildRecord> updateList = new ArrayList<>(100);
        List<MultiChildRecord> updateFailedList = new ArrayList<>(100);
        updateQueue.drainTo(updateList, 500);
        if (updateList.size() == 0) {
            return;
        }
        for (MultiChildRecord r : updateList) {
            if (!insertingMultiRecordSet.contains(buildSetKey(r))) {  //不在待插入列表中,说明已经入库
                example.createCriteria().andRecordIdEqualTo(r.getRecordId()).andCurrentIndexEqualTo(r.getCurrentIndex());
                int res = multiRecordMapper.updateByExampleSelective(r, example);
     //           System.err.println("修改的子请求数量: " + res + " , info => " + r);
                example.clear();
                if (res == 0) {//等于0表示数据库中没有该条日志,更新失败
                    //            log.info("id={}【子】请求还没有入库,不能更新", r.getRecordId());
                    updateFailedList.add(r);
                }
            } else {  //在的话,还没有入库,不能先更新
                //         log.info("id={}还没有入库,【子】请求不能更新", r.getRecordId());
                updateFailedList.add(r);
            }
        }
        if (updateFailedList.size() > 0) {      //如果有更新失败的日志,则放回到待更新日志
            //      log.warn("存在更新失败的【子】请求日志: " + Arrays.toString(updateFailedList.parallelStream().map(this::buildSetKey).collect(Collectors.toList()).toArray()));
            updateQueue.addAll(updateFailedList);
        }
    }
}
